package com.xl.flinkdemo.flinkState;

import com.xl.flinkdemo.entity.SensorReading;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.expressions.In;

import java.util.Collections;
import java.util.List;

/**
 * @Description: 键控状态
 * @Author: hxl
 * @Date: 2021/5/24 15:02
 * @param null:
 * @return: null
 **/
public class FlinkKeyedState {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    DataStream<String> stringDataStream =
        env.readTextFile("D:\\studyspace\\flinkdemo\\src\\main\\resources\\sensors\\input.txt");
    DataStream<SensorReading> dataStream = stringDataStream.map(new myMapFunction());


    //定义一个有状态的map操作，统计当前sensor数据个数
    DataStream<Integer> resultStream = dataStream.keyBy("id")
        .map(new MyKeyCountFunction());

    resultStream.print("result");

    env.execute();
  }

  /**
   * @Description: 自定义mapFunction
   * @Author: hxl
   * @Date: 2021/5/24 15:07
   **/
  public static class MyKeyCountFunction extends RichMapFunction<SensorReading,Integer> {

    //定义键控状态[]
    private ValueState<Integer> keyCountState ;

    @Override
    public Integer map(SensorReading sensorReading) throws Exception {
      // 读取键控状态
      Integer count = keyCountState.value();
      count++;
      //更改键控状态
      keyCountState.update(count);
      return count;
    }

    //必须要等生命周期open之后才能拿到 上下文
    @Override
    public void open(Configuration parameters) throws Exception {
      super.open(parameters);
      keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count",Integer.class));
    }
  }


  public static class myMapFunction implements MapFunction<String,SensorReading>{
    @Override
    public SensorReading map(String s) throws Exception {
      String[] fields = s.split(",");
      return new SensorReading(fields[0],new Long(fields[1]),new Double(fields[2]));
    }
  }



}
